package com.demo.flink.sql

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.catalog.hive.HiveCatalog

object DemoHive {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val settings = EnvironmentSettings.newInstance()
      .inBatchMode().build()
    val tEnv = StreamTableEnvironment.create(env, settings)
    val name = "myhive"
    val defaultDatabase = "default"
    val hiveConf = "/usr/local/hive/conf"
    val hiveCatalog = new HiveCatalog(name, defaultDatabase, hiveConf)
    tEnv.registerCatalog(name, hiveCatalog)
    tEnv.useCatalog(name)
    tEnv.useDatabase("dw")
    tEnv.executeSql("show tables").print()
    tEnv.executeSql("select * from dw_dim_customer").print()
  }
}
